package com.company.project.common.redismq;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.company.project.common.utils.StringUtil;
import com.company.project.entity.*;
import com.company.project.mapper.*;
import com.company.project.service.RedisService;
import com.company.project.service.TCallbakService;
import com.company.project.service.TCallbakTaskArcService;
import com.company.project.service.TCallbakTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.HttpClientErrorException;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;

@Slf4j
@Component
public class ZSetVer {
    public final static String RS_ZS_MQ_NS = "rzsm:";
    public final static String RS_ZS_MQ_ME = "consumerDelayMessage:";

    @Value("${spring.redis.host}")
    private String host;

    @Value("${spring.redis.port}")
    private int port;

    @Value("${spring.redis.password}")
    private String password;

    @Autowired
    private JedisPool jedisPool;

    @Resource
    public TCallbakTaskService tCallbakTaskService;
    @Resource
    public TCallbakTaskArcService tCallbakTaskArcService;
    @Resource
    public RedisService redisService;
    @Resource
    public TKaiguanMapper tKaiguanMapper;
    @Autowired
    public TSeatsMapper tSeatsMapper;
    @Autowired
    public TXiaohaoMapper xiaohaoMapper;
    @Resource
    public TCallbakTaskMapper tCallbakTaskMapper;
    @Resource
    private SysUserMapper sysUserMapper;

    @Autowired
    private TCallbakMapper callbakMapper;
    @Autowired
    private TCallbakLogMapper callbakLogMapper;
    @Autowired
    private TThjlmxMapper thjlmxMapper;
    @Autowired
    public TThrbMapper thrbMapper;
    @Autowired
    private THwtjMapper hwtjMapper;

    @Autowired
    private TMiddlenumMapper tMiddlenumMapper;
    @Autowired
    private TXiaohaoMapper tXiaohaoMapper;
    @Autowired
    private TWhbcsMapper tWhbcsMapper;
    @Autowired
    public TBacklistMapper backlistMapper;

    /*生产者，消息的发送，实际生产中，相关参数，
    比如订单信息，过期时间等应该传入，可以考虑将订单信息json化存入redis*/
    public void producer() {
        Jedis jedis = new Jedis(host, port);
        if (StringUtil.isNotBlank(password)){
            jedis.auth(password);
            jedisPool = new JedisPool(new JedisPoolConfig(), host, port, 2000, password);
        }
        try {
            jedis = jedisPool.getResource();
            for (int i = 0; i < 5; i++) {
                String order_id = "000000000"+i;
                double score = System.currentTimeMillis()+(i*1000);
                jedis.zadd(RS_ZS_MQ_NS+"orderId",score, order_id);
                System.out.println("生产订单: " + order_id + " 当前时间："
                        + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                System.out.println((3 + i) + "秒后执行");
            }
        } catch (Exception e) {
            throw new RuntimeException("生产消息失败！");
        } finally {
            jedis.close();
        }

    }

    @Async
    /*生产者，消息的发送，实际生产中，相关参数，
    比如订单信息，过期时间等应该传入，可以考虑将订单信息json化存入redis*/
    public void producer(String key,String value) {
        Jedis jedis = new Jedis(host, port);
        if (StringUtil.isNotBlank(password)){
            jedis.auth(password);
            jedisPool = new JedisPool(new JedisPoolConfig(), host, port, 2000, password);
        }
        try {
            jedis = jedisPool.getResource();
            double score = System.currentTimeMillis()+(1*1000);
            jedis.zadd(RS_ZS_MQ_NS+key,score, value);
            System.out.println("生产订单 key: " + key + " score："+score);
            System.out.println("生产订单: " + value + " 当前时间："
                    + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("生产消息失败！");
        } finally {
            jedis.close();
        }
    }

    //消费者，取订单
    public void consumerDelayMessage() {
        Jedis jedis = new Jedis(host, port);
        if (StringUtil.isNotBlank(password)){
            jedis.auth(password);
            jedisPool = new JedisPool(new JedisPoolConfig(), host, port, 2000, password);
        }
        try {
            jedis = jedisPool.getResource();
            while (true) {
                Set<String> order = jedis.zrangeByScore(RS_ZS_MQ_NS+"orderId", 0,
                        System.currentTimeMillis(), 0,1);
                if (order == null || order.isEmpty()) {
                    System.out.println("当前没有等待的任务");
                    try {
                        TimeUnit.MICROSECONDS.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    break;
                }
                String s = order.iterator().next();

                if (jedis.zrem(RS_ZS_MQ_NS+"orderId", s)>0) {
                    /*业务处理*/
                    System.out.println(s);
                }
            }

        } catch (Exception e) {
            throw new RuntimeException("消费消息失败！");
        } finally {
            jedis.close();
        }
    }

    // 心跳检测
//    @Scheduled(cron = " */30 * * * * ?")
    public void run() {
        String key = "CallBack";
        String s = redisService.get(RS_ZS_MQ_ME+key);
        if (StringUtil.equals("false",s)) {
            consumerDelayMessage(key);
        }
    }
    boolean flsh = false;
    //消费者，取订单
    public void consumerDelayMessage(String key) {
        long start = System.currentTimeMillis();

        Jedis jedis = new Jedis(host, port);
        if (StringUtil.isNotBlank(password)){
            jedis.auth(password);
            jedisPool = new JedisPool(new JedisPoolConfig(), host, port, 2000, password);
        }

        jedis = jedisPool.getResource();
        List<TCallbakTaskEntity> list = new ArrayList<>();

        // 创建一个固定大小的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        try {
            //总条数
            Long zcard = jedis.zcard(RS_ZS_MQ_NS+key);
            System.out.println("consumerDelayMessage  key：{ "+key+" } 总条数：{ "+zcard+" } 个");
            if (zcard > 0) {
                flsh = true;
            }
            while (flsh) {
                //每次获取条数
                int count = 200;
                if (zcard < 200){
                    count = Math.toIntExact(zcard);
                }
                Set<String> order = jedis.zrangeByScore(RS_ZS_MQ_NS+key, 0,
                        System.currentTimeMillis(), 0,count);
                System.out.println("order: ："+order.toString());
                if (order == null || order.isEmpty()) {
                    System.out.println("当前没有等待的任务---key："+key);
                    flsh = false;
                } else {
                    System.out.println("runCallBackDscy---处理条数："+order.size());
                    System.out.println("runCallBackDscy---order.toString()："+order.toString());
                    for (String s : order) {
                        System.out.println("runCallBackDscy---orders 遍历："+s);
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        TCallbakTaskEntity javaObject = JSON.toJavaObject(jsonObject, TCallbakTaskEntity.class);
                        if (StringUtil.isNotBlank(javaObject.getBindId()))
                        list.add(javaObject);
                    }
                    if (list.size() > 0) {
                        List<Future<JSONObject>> futures = new ArrayList<>();
                        List<TCallbakTaskEntity> finalList = list;
                        System.out.println("runCallBackDscy---处理条数："+list.size());
//                        runCallBackDscy(list);
                        Callable<JSONObject> callable = new Callable<JSONObject>() {
                            @Override
                            public JSONObject call() throws Exception {
                                JSONObject jsonObject = runCallBackDscy(finalList);
                                return jsonObject;
                            }
                        };

                        Future<JSONObject> future = executorService.submit(callable);
                        futures.add(future);
//                        for (Future<JSONObject> fut : futures) {
////                             注意：get()方法会阻塞，直到任务完成
//                        JSONObject square = fut.get();
//                            //处理数据
//                            tCallbakTaskService.insertByJson(square);
//                        System.out.println(square);
//                        }
                    }
                    for (String s : order) {
                        System.out.println(" zrem  ---orders 遍历："+s);
                        //移除有序集合中的一个或多个成员
                        Long zrem = jedis.zrem(RS_ZS_MQ_NS + key, s);
                        System.out.println(" zrem  ---orders 删除成功");
                    }
                    list = new ArrayList<>();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("消费消息失败！");
        } finally {
            jedis.close();
        }
        // 关闭线程池
        executorService.shutdown();
        // 等待线程池中的任务都执行完毕
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                // 线程池没有在给定的时间内终止，可以选择取消它
                executorService.shutdownNow();
            }
        } catch (InterruptedException ie) {
            // 当前线程在等待过程中被中断
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }

        // 所有线程都已完成，继续主线程的其他操作（如果有的话）
        System.out.println("All threads have finished.");
        long end = System.currentTimeMillis();
        System.out.println("consumerDelayMessage耗时：{ "+ ( end - start)+" } ms");
    }

    JSONObject runCallBackDscy(List<TCallbakTaskEntity> list) {
        System.out.println("list 总数量：{ "+ list.size()+" } ms");
        long start = System.currentTimeMillis();
        //入库
        String  s =  tCallbakTaskArcService.runCallBack(list);
        JSONObject jsonObject = JSON.parseObject(s);
        long end = System.currentTimeMillis();
        log.info("入库 runCallBack总耗时：{} ms", end - start);
        return jsonObject;
    }
    public void consumerDelayMessageData(String key) {
        long start = System.currentTimeMillis();
        Jedis jedis = new Jedis(host, port);
        if (StringUtil.isNotBlank(password)){
            jedis.auth(password);
            jedisPool = new JedisPool(new JedisPoolConfig(), host, port, 2000, password);
        }
        try {
            jedis = jedisPool.getResource();
            //总条数
            Long zcard = jedis.zcard(RS_ZS_MQ_NS+key);
            System.out.println("consumerDelayMessage  key：{ "+key+" } 总条数：{ "+zcard+" } 个");
            if (zcard > 0) {
                flsh = true;
                redisService.set(RS_ZS_MQ_ME+key,"true");
            }
            while (flsh) {
                //每次获取条数
                int count = 200;
                if (zcard < 200){
                    count = Math.toIntExact(zcard);
                }
                Set<String> order = jedis.zrangeByScore(RS_ZS_MQ_NS+key, 0,
                        System.currentTimeMillis(), 0,count);
                if (order == null || order.isEmpty()) {
                    System.out.println("当前没有等待的任务---key："+key);
                    redisService.set(RS_ZS_MQ_ME+key,"false");
                    flsh = false;
                } else {
                    //处理数据
                    tCallbakTaskService.insertBykeys(key,order);
                    for (String s : order) {
                        //移除有序集合中的一个或多个成员
                        Long zrem = jedis.zrem(RS_ZS_MQ_NS + key, s);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("消费消息失败！");
        } finally {
            jedis.close();
        }
        long end = System.currentTimeMillis();
        System.out.println("consumerDelayMessage耗时：{ "+ ( end - start)+" } ms");
    }

}
